﻿//#include <arpa/inet.h>
#include "SLT_log.h"
#include "SLT_tcpServerImpl.h"
#include "SLT_tcpSenderImpl.h"

#define  _LOG_SLTTCPSENDER_DEBUG(_id,_format,...)  SLT_LOG(1,SLT_DEBUG"[tcpsender](%u)" _format,_id, ##__VA_ARGS__)
#define  _LOG_SLTTCPSENDER_INFO(_id,_format,...)  SLT_LOG(1,SLT_INFO,"[tcpsender](%u)" _format,_id, ##__VA_ARGS__)
#define  _LOG_SLTTCPSENDER_ERR(_id,_format,...)  SLT_LOG(1,SLT_ERROR,"[tcpsender](%u)" _format,_id, ##__VA_ARGS__)

//tcp连接协议
//协议格式:  长度（int)| 命令（short)| 目的sessiondid| 源sessionid | 内容
#pragma pack(1)
struct SLTTcpProtocolHeadSt
{
	uint32_t  m_len;
	uint16_t  m_cmd;
	uint32_t  m_dstid;
	uint32_t  m_srcid;
	
	void toLocal() {
		m_len = SLT_ntohl(m_len);
		m_dstid = SLT_ntohl(m_dstid);
		m_srcid = SLT_ntohl(m_srcid);
		m_cmd = SLT_ntohs(m_cmd);
	}
	void toNet() {
		m_len = SLT_htonl(m_len);
		m_dstid = SLT_htonl(m_dstid);
		m_srcid = SLT_htonl(m_srcid);
		m_cmd = SLT_htons(m_cmd);
	}
	std::string toString() {
		char buf[128] = {0};
		snprintf(buf, sizeof(buf), "%u,%u,%u,%u", m_len, m_cmd, m_dstid, m_srcid);
		return std::string(buf);
	}
};
//代理连接建立头协议：
struct SLTTcpProxyProtocolHeadSt
{
	uint32_t  m_len;
	uint16_t  m_cmd;

	void toLocal() {
		m_len = SLT_ntohl(m_len);
		m_cmd = SLT_ntohs(m_cmd);
	}
	void toNet() {
		m_len = SLT_htonl(m_len);
		m_cmd = SLT_htons(m_cmd);
	}
	std::string toString() {
		char buf[64] = { 0 };
		snprintf(buf, sizeof(buf), "%u,%u", m_len, m_cmd);
		return std::string(buf);
	}
};
#pragma pack()

SLT_tcpRecvMsgCb_type  SLT_tcpSenderProtocolImpl::m_precvMsgCBFun;
SLT_tcpSenderProtocolImpl::SLT_tcpSenderProtocolImpl(SLT_tcpServerFrameImpl* _pser)
{
	m_tcpServer = _pser;
	m_lastDataTime = SLT_getSteadyTimeMs();
}

void SLT_tcpSenderProtocolImpl::connect(SLTcpClientPtr &_client,const std::string& _rip, int _rport)
{
	//std::unique_lock<std::mutex> a(m_tcpClientLock);  //因为目前是同步连接事件。造成死锁
	if (_sender_work_state_init != m_workState) {
		_LOG_SLTTCPSENDER_ERR(m_id, "repeat connect error");
		return;
	}
	
	m_rip = SLT_ip2num(_rip.c_str());
	m_rStrIp = _rip;
	m_rport = _rport;
	m_workMode = _sender_work_mode_cli;

	if(! m_tcpClient){
		m_tcpClient = _client;
		m_tcpClient->bindCB(shared_from_this());
	}
	
	_LOG_SLTTCPSENDER_ERR(m_id, "start connect :%X,%d", m_rip, m_rport);
	m_workState = _sender_work_state_conning;
	m_reconnectDiffTime = -1;
	m_lastConnectTime = SLT_getSteadyTimeMs();
	m_tcpClient->connect2(_rip, _rport);
}
void  SLT_tcpSenderProtocolImpl::recvConnect(SLTcpClientPtr& _cliPtr)
{
	_cliPtr->bindCB(shared_from_this());
	m_tcpClient = _cliPtr;

	m_workState = _sender_work_state_ok;
	const slIpAddr& sladdr = m_tcpClient->getRemoteAddr();
	m_rip = sladdr.m_ip;
	m_rport = sladdr.m_port;
	
}
SLT_TCPSenderPtr SLT_tcpSenderProtocolImpl::getSenderApt()
{
	return std::make_shared<SLT_tcpSenderProtocolApt>(shared_from_this());
}
std::string SLT_tcpSenderProtocolImpl::getKey()
{
	std::string strip = SLT_ip2str(m_rip);
	char strkey[128];
	snprintf(strkey,sizeof(strkey), "%s_%u", strip.c_str(), m_rport);
	return std::string(strkey);
}
bool SLT_tcpSenderProtocolImpl::SLF_postMsg(uint32_t _SessID, int msgid, uint32_t _srcID,const SLF_sendMemBasePtr& _data)
{
	m_lastDataTime = SLT_getSteadyTimeMs();

	if (m_workState != _sender_work_state_ok) {
		_LOG_SLTTCPSENDER_ERR(m_id,"send error. connect failed");
		return false;
	}

	int datalen = _data->getLen();
	int membuflen = sizeof(SLTTcpProtocolHeadSt) + datalen;
	char* pmembuf = new char[membuflen];

	SLTTcpProtocolHeadSt* prespHead = (SLTTcpProtocolHeadSt*)pmembuf;
	prespHead->m_cmd = msgid;
	prespHead->m_len = membuflen;
	prespHead->m_dstid = _SessID;
	prespHead->m_srcid = _srcID;

	prespHead->toNet();
	if (_data->getData()) {
		memcpy(pmembuf + sizeof(SLTTcpProtocolHeadSt), _data->getData(), datalen);
	}

	int ret = -1;
	m_tcpClientLock.lock();
	if (m_tcpClient) {
		ret = m_tcpClient->send2(pmembuf, membuflen);
	}
	m_tcpClientLock.unlock();

	delete pmembuf;
	if ( ret >= 0) {
		return true;
	}
	return false;
}
void SLT_tcpSenderProtocolImpl::TcpClient_onRecv(char* _pdata, int _len)
{
	m_precvAllMem.append(_pdata, _len);
	while (1) {
		if (0 != recvSplitCmd()) {
			break;
		}
	}
}
int SLT_tcpSenderProtocolImpl::recvSplitCmd()
{
	if (m_precvAllMem.getLen() < sizeof(SLTTcpProtocolHeadSt)) {
		return -1;
	}
	m_lastDataTime = SLT_getSteadyTimeMs();  //解决代理重复连接。造成空消息更新了时间

	SLTTcpProtocolHeadSt localhead;
	memcpy(&localhead, m_precvAllMem.get(), sizeof(SLTTcpProtocolHeadSt));
	//std::string tempStr;
	//SLT_AscToHex2((char*)&localhead, sizeof(proxyClientProtoHead), tempStr);
	//_COMMPROTOCOL_LOG_INFO(0, "recv newhead:%s", tempStr.c_str());
	localhead.toLocal();

	uint32_t packetLen = localhead.m_len;
	if (packetLen > m_precvAllMem.getLen()) {
		//命令不完整，等待
		//_CACTCPPROT_LOG_DEBUG(0,"tcp recv cmd not full.%d,%d", localhead.m_len, m_precvAllMem.getLen());
		return -3;
	}
	
	//SLF_sendMemBasePtr mbufPtr = std::make_shared<SLTTcpProtocolMemBase>(sizeof(SLTTcpProtocolHeadSt));
	SLF_sendMemBasePtr mbufPtr(new SLTTcpProtocolMemBase(sizeof(SLTTcpProtocolHeadSt)));
	if (packetLen == m_precvAllMem.getLen()) {
		mbufPtr->m_mem = m_precvAllMem;
		m_precvAllMem.reset();	}
	else {
		mbufPtr->m_mem.add(m_precvAllMem.get(), packetLen);
		m_precvAllMem.removeHead(packetLen);  //移动内存
	}
	
	if (0 == localhead.m_dstid) {
		if (!m_precvMsgCBFun) {
			_LOG_SLTTCPSENDER_ERR(m_id, "recv data error . datacb null");
		}
		else {
			SLF_SessionBasePtr  dstSessionSptr;
			m_precvMsgCBFun(0, localhead.m_cmd, localhead.m_srcid, mbufPtr, shared_from_this(), dstSessionSptr);
			if (dstSessionSptr) {
				SLF_pushEvent(dstSessionSptr->getSessID(), localhead.m_cmd, localhead.m_srcid, mbufPtr, shared_from_this());
			}
		}
		
	}
	else {
		if(false == SLF_pushEvent(localhead.m_dstid, localhead.m_cmd, localhead.m_srcid, mbufPtr, shared_from_this())){
			if (!m_precvMsgCBFun) {
				_LOG_SLTTCPSENDER_ERR(m_id, "recv data error . datacb null");
			}
			else {
				SLF_SessionBasePtr  dstSessionSptr;
				m_precvMsgCBFun(localhead.m_dstid, localhead.m_cmd, localhead.m_srcid, mbufPtr, shared_from_this(), dstSessionSptr);
				if (dstSessionSptr) {
					SLF_pushEvent(dstSessionSptr->getSessID(), localhead.m_cmd, localhead.m_srcid, mbufPtr, shared_from_this());
				}
			}
		}
	}
	return 0;
}
void SLT_tcpSenderProtocolImpl::TcpClient_onConnect(tcp_error_type _errType)
{
	std::unique_lock<std::mutex> a(m_tcpClientLock);
	//主动连接行为
	if (tcp_error_type_ok == _errType) {
		//尝试发送缓存
		if (_sender_work_state_conning != m_workState) {
			_LOG_SLTTCPSENDER_ERR(m_id, "connect stat error:%X,%d", m_rip, m_rport);
			if (m_tcpClient) {  //这个逻辑正常应该不会触发
				m_tcpClient->close2();
			}
			return;
		}
		_LOG_SLTTCPSENDER_ERR(m_id, "connect succ:%X,%d", m_rip, m_rport);
		m_precvAllMem.clear();
		m_workState = _sender_work_state_ok;

		m_reconnectDiffTime = -1;
		connectSuccNotify();
	}
	else
	{
		if (_sender_work_state_conning != m_workState) {
			_LOG_SLTTCPSENDER_ERR(m_id, "connect stat error:%X,%d", m_rip, m_rport);
			return;
		}
		_LOG_SLTTCPSENDER_ERR(m_id, "connect failed:%X,%d", m_rip, m_rport);
		m_workState = _sender_work_state_init;
		//连接失败，等待5s重连
		m_reconnectDiffTime = 5000;
	}
}
void SLT_tcpSenderProtocolImpl::TcpClient_onError(tcp_error_type _type)
{
	std::unique_lock<std::mutex> a(m_tcpClientLock);

	_LOG_SLTTCPSENDER_ERR(m_id, "tcp error, radd(%X,%d),type:%d", m_rip, m_rport, _type);
	if (m_workMode == _sender_work_mode_cli) {
		//连接成功后失败，1s后重连
		if (_sender_work_state_stop != m_workState) {
			m_workState = _sender_work_state_init;
			m_lastConnectTime = SLT_getSteadyTimeMs();
			if(tcp_error_type_localclose == _type){
				m_reconnectDiffTime = 5000;
			}else{
				m_reconnectDiffTime = 1000;
			}
			
		}
	}
	else {
		m_workState = _sender_work_state_stop;
		m_tcpClient.reset();
	}
}
void SLT_tcpSenderProtocolImpl::_closeSender()
{
	std::unique_lock<std::mutex> a(m_tcpClientLock);
	m_workState = _sender_work_state_init;
	if (m_tcpClient) {
		m_tcpClient->close2();
	}
	//关闭后触发关闭事件，会重新建立连接
}
void SLT_tcpSenderProtocolImpl::checkState()
{
	// if (0 == m_usernum) {
	// 	//没有人使用且没有收发数据超时删除： 3分钟
	// 	uint64_t tnow = SLT_getSteadyTimeMs();
	// 	uint64_t tdiff = tnow - m_lastDataTime;
	// 	if (tdiff > 30000) {
	// 		m_lastDataTime = SLT_getSteadyTimeMs();
	// 		std::unique_lock<std::mutex> a(m_tcpClientLock);  //加锁
	// 		if (0 == m_tcpServer->tryDelSender(getKey())) {
	// 			_LOG_SLTTCPSENDER_ERR(m_id, "sender timeout del");
	// 			m_workState = _sender_work_state_stop;
	// 			if (m_tcpClient) {
	// 				m_tcpClient->close2();
	// 				m_tcpClient.reset();
	// 			}
	// 		}
	// 	}
	// }
	if(0 == m_usernum){
		if (0 == m_tcpServer->tryDelSender(getKey())) {
			std::unique_lock<std::mutex> a(m_tcpClientLock);  //加锁
			_LOG_SLTTCPSENDER_ERR(m_id, "sender del");
			m_workState = _sender_work_state_stop;
			if (m_tcpClient) {
				m_tcpClient->close2();
				m_tcpClient.reset();
			}
			return;
		}
	}
	
	if (-1 != m_reconnectDiffTime) {
		uint64_t tnow = SLT_getSteadyTimeMs();
		if (tnow - m_lastConnectTime > (uint64_t)m_reconnectDiffTime) {
			connect(m_tcpClient,m_rStrIp, m_rport);
		}
	}
}
void SLT_tcpSenderProtocolImpl::delref() 
{
	//没有立即删除，防止删除和connect多线程问题。connect 目前无法加锁（因为是同步连接回调，回调中有锁）
	 int ret = --m_usernum;
	//  if(0 == ret){
	// 	if (0 == m_tcpServer->tryDelSender(getKey())) {
	// 		std::unique_lock<std::mutex> a(m_tcpClientLock);  //加锁
	// 		_LOG_SLTTCPSENDER_ERR(m_id, "sender del");
	// 		m_workState = _sender_work_state_stop;
	// 		if (m_tcpClient) {
	// 			m_tcpClient->close2();
	// 			m_tcpClient.reset();
	// 		}
	// 	}
	//  }
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
std::string SLT_TCPProxyTransSenderImpl::getKey()
{
	char strkey[64];
	snprintf(strkey,sizeof(strkey), "%u_proxysender", m_id);
	return std::string(strkey);
}
void SLT_TCPProxyTransSenderImpl::connectSuccNotify()
{
	//发送握手协议
	m_workState = _sender_work_state_handshaking;
	m_handsharkeTimer.start();

	CProxyProHandshake  hand;
	hand.ht_timeout_ms = m_timeout_ms;
	hand.remote_agent_v2v_number.no_pref1 = m_prefix1;
	hand.remote_agent_v2v_number.no_pref2 = m_prefix2;
	hand.remote_agent_v2v_number.no_pref3 = m_prefix3;
	hand.remote_agent_v2v_number.dev_no = m_devno;
	hand.remote_agent_v2v_number.dev_sub_no = m_subno;
	//hand.svr_id = "0x9998";
	hand.svr_id = m_svrid;
	hand.encrypt = m_encrypt;

	slt_uuid_t uuid;
	slt_uuid_create(&uuid);
	hand.sessionid = slt_uuid_to_string(&uuid);

	std::string strJson;
	aigc::JsonHelper::ObjectToJson(hand, strJson);
	SLT_memBuf membuf;
	membuf.resize(sizeof(SLTTcpProxyProtocolHeadSt) + strJson.length() + 1);

	SLTTcpProxyProtocolHeadSt* prespHead = (SLTTcpProxyProtocolHeadSt*)membuf.bufAddr();
	prespHead->m_cmd = 1;
	prespHead->m_len = membuf.getSize();
	prespHead->toNet();
	memcpy(membuf.bufAddr() + sizeof(SLTTcpProxyProtocolHeadSt), strJson.c_str(), strJson.length()+1);
	_LOG_SLTTCPSENDER_ERR(m_id,"proxy connect send:%s", strJson.c_str());
	m_tcpClient->send2(membuf.bufAddr(), membuf.m_size);
}
int SLT_TCPProxyTransSenderImpl::recvSplitCmd()
{
	//代理tcp连接建立成功后，先发送连接信息。接收到应答，连接建立成功，再接收到的消息就是应用层的消息了（代理不再对tcp上的流信息进行处理)。通过SLT_tcpSenderProtocolImpl 协议处理
	if (m_workState == _sender_work_state_ok) {
		return SLT_tcpSenderProtocolImpl::recvSplitCmd();
	}
	if (m_workState != _sender_work_state_handshaking) {
		_LOG_SLTTCPSENDER_ERR(m_id,"proxy connect resp state error!!!!!:%d", m_workState);
		return -1;
	}
	if (m_precvAllMem.getLen() < sizeof(SLTTcpProxyProtocolHeadSt)) {
		return -1;
	}
	SLTTcpProxyProtocolHeadSt localhead;
	memcpy(&localhead, m_precvAllMem.get(), sizeof(SLTTcpProxyProtocolHeadSt));
	localhead.toLocal();

	uint32_t packetLen = localhead.m_len;
	if (packetLen > m_precvAllMem.getLen()) {
		return -3;
	}
	//与代理连接成功，从接收缓冲移除应答命令

	CProxyProHandshake_resp resp;
	std::string errJson;
	if (!aigc::JsonHelper::JsonToObject2(resp, m_precvAllMem.get() + sizeof(SLTTcpProxyProtocolHeadSt))) {
		_LOG_SLTTCPSENDER_ERR(m_id, "proxy connect resp json error:%s, json:%s", errJson.c_str(), m_precvAllMem.get() + sizeof(SLTTcpProxyProtocolHeadSt));
		_closeSender();
		return -4;
	}

	if (resp.ret != 0) {
		_LOG_SLTTCPSENDER_ERR(m_id, "proxy connect resp error,code:%d,msg:%s,sessionid:%s", resp.ret, resp.ret_msg.c_str(), resp.sessionid.c_str());
		_closeSender();
		return -4;
	}
	
	_LOG_SLTTCPSENDER_INFO(m_id, "proxy connect resp ok:%s", m_precvAllMem.get() + sizeof(SLTTcpProxyProtocolHeadSt));
	m_precvAllMem.removeHead(packetLen);

	m_workState = _sender_work_state_ok;
	return 0;
}

